#include "RelatedRecomm_types.h"
#include "RelatedRecomm.h"

#include "service_arch/service_arch.h"
#include "service_arch/service_base_impl.h"
// #include "database/simple_db_pool.h"
#include "base/logging.h"

#include <stdio.h>
#include <iostream>
#include <map>

#include <boost/thread/thread.hpp>
#include <boost/algorithm/string.hpp>
#include <boost/lexical_cast.hpp>

#include <thrift/Thrift.h>

#include <thrift/transport/TServerTransport.h>
#include <thrift/transport/TServerSocket.h>
#include <thrift/transport/TBufferTransports.h>

#include <thrift/protocol/TProtocol.h>
#include <thrift/protocol/TBinaryProtocol.h>

#include <thrift/concurrency/ThreadManager.h>
#include <thrift/server/TThreadPoolServer.h>

using namespace std;
using namespace boost;
using namespace apache::thrift;

namespace jztzp {

class RelatedRecommHandler : public RelatedRecommIf, public ServiceBaseImpl {
 public:
  RelatedRecommHandler() {
    srand(time(NULL));
    bool ret = Reload("./article_features.txt");
    LOG_INFO("Load default feature file ret=" << ret);
  }

  virtual bool Reload(const string& data_file) {
    FILE * data = fopen(data_file.c_str(), "r");
    if (!data) {
      // LOG_WARN("load from stat file " << data_file_ << " error");
      return false;
    }

    static const int kMaxLine = 1024;
    char line[kMaxLine];
    line[kMaxLine - 1] = 0;
    std::vector<std::string> strs;
    while(fgets(line, kMaxLine - 1, data)) {
      int article_id = 0;
      LOG_DEBUG("load " << line);
      boost::split(strs, line, boost::is_any_of(" \t\r\n"), boost::token_compress_on);
      if (strs.size() < 3) {
        continue;
      }
      try {
        article_id = boost::lexical_cast<int>(strs[0]);
      } catch (boost::bad_lexical_cast & e) {
        // LOG_WARN("load bad stat data : " << line);
        continue;
      }
      if (article_id <= 0) {
        continue;
      }
      Article article;
      article.id = article_id;
      for(int i = 1; i < strs.size() - 1; i += 2) {
        double weight = boost::lexical_cast<double>(strs[i + 1]);
        article.features[strs[i]] = weight;
      }
      AddArticle(article);
    }
    fclose(data);
    return true;
  }

  virtual bool AddArticle(const Article & article) {
    boost::unique_lock<boost::shared_mutex> lock(shared_mutex_);

    id_featyre_idx_[article.id] = article;
    map<string, double>::const_iterator it = article.features.begin();
    for(; it != article.features.end(); ++it) {
      // featyre_id_idx_[it->first].insert(it->second);
      if (it->first.size() <= 1) {
        LOG_WARN("bad feature " << it->first.size());
        continue;
      }
      featyre_id_idx_[it->first].insert(article.id);
    }
    return true;
  }
  virtual bool RemoveArticle(const int32_t id) {
    boost::unique_lock<boost::shared_mutex> lock(shared_mutex_);
    id_featyre_idx_.erase(id);
    return true;
  }

  void GetRelated(std::vector<int32_t> & related, const int32_t id) {
    map<int32_t, double> related_score;
    boost::shared_lock<boost::shared_mutex> lock(shared_mutex_);
    map<int32_t, Article>::const_iterator it = id_featyre_idx_.find(id);
    if (it == id_featyre_idx_.end()) {
      return;
    }

    const map<string, double> & features = it->second.features;
    map<string, double>::const_iterator fit = features.begin();
    for(; fit != features.end(); ++fit) {
      map<string, set<int32_t> >::const_iterator related_it = featyre_id_idx_.find(fit->first);
      if (related_it != featyre_id_idx_.end()) {
        LOG_DEBUG("feature " << fit->first.c_str() << " related count=" << related_it->second.size());

        set<int32_t>::const_iterator id_it = related_it->second.begin();
        for (; id_it != related_it->second.end(); ++id_it) {
          if (*id_it == id) {
            continue;
          }
          LOG_DEBUG("feature " << fit->first.c_str() << " related id=" << *id_it);
          map<int32_t, double>::iterator score_it = related_score.find(*id_it);
          if (score_it != related_score.end()) {
            related_score[*id_it] += fit->second; // fit->first.size(); // 1.0; 
          } else {
            related_score[*id_it] = fit->second; // fit->first.size(); // 1.0;
          }
        }
      }
    }
    
    SelectRelated(related_score, &related);
  }
 private:
  void GetRandomIds(const vector<int32_t> & ids, int max_count, set<int32_t> * ret) {
    set<int32_t> candidates;
    for(int i = 0; i < ids.size(); ++i) {
      candidates.insert(ids[i]);
    }
    if (max_count < 1) {
      max_count = 1;
    }
    while(candidates.size() > 0 && ret->size() < max_count) {
      int step = rand() % candidates.size();
      set<int32_t>::iterator it = candidates.begin();
      std::advance(it, step);
      ret->insert(*it);
      candidates.erase(it);
    }
  }

  double CalcTotalWeight(const map<double, vector<int32_t> > & sorted_related) {
    double total_weight = 0.0;
    map<double, vector<int32_t> >::const_iterator it = sorted_related.begin();
    for(; it != sorted_related.end(); ++it) {
      total_weight += it->first;
    }
    return total_weight;
  }

  void SelectRelated(const map<int32_t, double> & candidates, vector<int32_t> * selected) {
    map<double, vector<int32_t> > sorted_related;
    map<int32_t, double>::const_iterator it = candidates.begin();
    for(; it != candidates.end(); ++it) {
      sorted_related[it->second].push_back(it->first);
      LOG_DEBUG("Related score " << it->second << " aid=" << it->first << "/" << sorted_related[it->second].size());
    }

    double total_weight = CalcTotalWeight(sorted_related);
    int max_selected = 12;

    for(map<double, vector<int32_t> >::reverse_iterator rit = sorted_related.rbegin();
        rit != sorted_related.rend();
        ++rit) {
      set<int32_t> rand_ids;
      GetRandomIds(rit->second, int((max_selected - selected->size()) * rit->first / total_weight), &rand_ids);
      for(set<int32_t>::const_iterator sit = rand_ids.begin(); sit != rand_ids.end(); ++ sit) {
        selected->push_back(*sit);
      }
      total_weight -= rit->first;
    }
  }

  map<int32_t, Article> id_featyre_idx_;
  map<string, set<int32_t> > featyre_id_idx_;
  boost::shared_mutex shared_mutex_;
};

}

int main(int argc, char **argv) {
  LOG_INIT("./related_recomm_service.log", "INFO");
  int port = 9091;
  cout << "listens on port " << port << endl;

  using namespace jztzp;
  ThriftService<RelatedRecommHandler, RelatedRecommProcessor> service;
  service.Start(port);
  return 0;
}

